package job

import (
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/eventkit/dxc-server/repo"
	"git.multiverse.io/eventkit/kit/contexts"
	"git.multiverse.io/framework/log"
	"strings"
)

func StartSyncUnfinishedRecordsFromStorageJob() (err error) {
	log.Infos("Start sync. unfinished records from storage ...")
	defer func() {
		if nil == err {
			log.Infos("Successfully synchronized unfinished records from storage!")
		} else {
			log.Errorsf("Failed to synchronize unfinished records from storage, error=%++v", err)
		}
	}()

	repoType := config.Cfg.Repo.Type
	if repoType != repo.Hybrid_With_Redis_Storage && repoType != repo.Hybrid_With_Hazelcast_Storage {
		log.Debugsf("Repository is %s, no need to get abnormal transactions", repoType)
		return nil
	}
	log.Debugsf("Start Down Recover Job: need to get transactions from redis")
	txnRepo := repo.RepositoryInstance()
	hybridRepo := txnRepo.(*repo.HybridStorage)
	redundancyRepo := hybridRepo.GetRedundancyRepo()
	keys, err := redundancyRepo.LoadRootsFromPersistenceRepo(config.Cfg.Job.BatchSyncSize)
	if err != nil {
		return err
	}
	if len(keys) == 0 {
		log.Debugsf("Start Down Recover Job: redis no transactions")
		return nil
	}
	rootXIds := make([]string, 0)
	for _, key := range keys {
		rootXIds = append(rootXIds, strings.Split(key, ":")[1])
	}
	var count int
	if len(rootXIds) < 100 {
		count = 1
	} else if len(rootXIds) < 500 {
		count = 3
	} else if len(rootXIds) < 1000 {
		count = 5
	} else {
		count = 10
	}
	size := len(rootXIds) / count
	for i := 0; i < count; i++ {
		var ids []string
		if i == count-1 {
			ids = rootXIds[i*size:]
		} else {
			ids = rootXIds[i*size : (i+1)*size]
		}
		go downRecoverWorker(i, ids, hybridRepo)
	}
	return nil
}

func downRecoverWorker(workerId int, rootXIds []string, hybridRepo *repo.HybridStorage) {
	log.Debugsf("Start Down Recover Job: worker:[%d] get transactions from redis, rootXIds:[%++v]", workerId, rootXIds)
	memoryRepo := hybridRepo.GetMemoryRepo()
	redundancyRepo := hybridRepo.GetRedundancyRepo()
	for _, rootXId := range rootXIds {
		root, err := redundancyRepo.FindAllBranchByRootXid(rootXId)
		if err != nil {
			log.Errorsf("Start Down Recover Job: worker:[%d] FindAllBranchByRootXid From Redis fail, err:%s", workerId, err)
			continue
		}
		root.HasExpireTime = true
		_, err = memoryRepo.Recovery(root)
		if err != nil {
			log.Errorsf("Start Down Recover Job: worker:[%d] Memory Repository Recover fail, err:%s", workerId, err)
			continue
		}
		// find root transaction, parentXid is traceId
		var traceId string
		for _, branch := range root.Branches {
			if branch.IsRoot() {
				traceId = branch.ParentXid
			}
		}
		log.Debugsf("Start Down Recover Job: worker:[%d] Push Queue Recover rootXid:[%s]", workerId, rootXId)
		spanCtx := contexts.BuildSpanContexts(contexts.TraceID(traceId),
			contexts.ParentSpanID(rootXId),
			contexts.TimeoutMilliseconds(30000),
		)
		pair := models.Pair{
			RootXid:  rootXId,
			SpanCtx:  spanCtx,
			RootType: models.UnfinishedRecordsRecover,
			Type:     models.UnfinishedRecordsRecover,
		}
		models.Queue.Push(pair)
	}
}
